package edu.csl.study.spark.basic

import java.sql.Timestamp
import java.text.SimpleDateFormat
import java.util.Date

import org.apache.spark.sql.streaming.{StreamingQuery, Trigger}
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

object StructuredStreaming_window {
  val rootDir = System.getProperty("user.dir")+ "\\testFile\\"
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder().appName("StructuredStreaming").master("local[2]").getOrCreate()
    spark.sparkContext.setLogLevel("WARN")

    val lines:DataFrame = source_Socket(spark)
    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
    import spark.implicits._
    //转换成(2019-03-08 12:00:00,"dog") 格式 其中第一个元素为
    val words:DataFrame = lines.as[String].map(s =>{
       val arr = s.split(",")
       val date:Date = sdf.parse(arr(0))
       (new Timestamp(date.getTime),arr(1))
    }).toDF("ts","word")
    //必须导入这个，window才能编译
    import org.apache.spark.sql.functions._
    val workCounts  = words.groupBy(window($"ts","10 minutes","2 minutes"),$"word").count()
    /**
     * 3中输出模式：
     *  Complete表示结果全量输出,这种模式必须存在聚合操作
     *  Append 追加模式   非Watermark模式下，不能使用任何聚合操作
     *  Update 更新模式 Spark 2.1.1之后支持，如果查询结果中没有聚合操作，那么等同于Append模式。
     */
    val query:StreamingQuery = workCounts.writeStream.outputMode("complete")
      .trigger(Trigger.ProcessingTime(0))//0表示处理完当前批次，立即处理下一批
      .format("console")
      .start()

    query.awaitTermination();

  }

  /**
   * socket来源
   *   nc -lk 8091 测试
   * @param spark
   * @return
   */
  def source_Socket(spark:SparkSession):DataFrame = {
    val source:DataFrame = spark.readStream.format("socket")
      .option("host","192.168.100.20")
      .option("port",8091)
      .load();
    source
  }

}
